

package com.qiyu.mq.core.consumer;

import com.alibaba.fastjson.JSON;
import com.qiyu.mq.core.annotation.MqConsumer;
import com.qiyu.mq.core.config.MqConsumerConfig;
import com.qiyu.mq.core.config.RocketMQProperties;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MQConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.hook.ConsumeMessageHook;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author  idea
 * @version V1.0
 * @date 2020/8/10
 */
public class DefaultConsumerFactory implements ConsumerFactory,ApplicationContextAware, InitializingBean, DisposableBean, BeanPostProcessor {

    private ApplicationContext applicationContext;

    private String nameServers;

    private MqConsumerConfig consumerConfig;

    //todo 目前只兼容顺序消费
    private Map<String, ConcurrentlyConsumer> topicConsumerMap = new HashMap<>();


    public DefaultConsumerFactory(String nameServers, MqConsumerConfig consumerConfig,String env) {
        //目前只支持测试环境的构建
        if("test".equals(env)){
            this.nameServers = nameServers;
            this.consumerConfig = consumerConfig;
        }
    }

    @Override
    public void destroy() throws Exception {

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Map<MessageModel, DefaultMQPushConsumer> defaultMQConsumerMap = new HashMap<MessageModel, DefaultMQPushConsumer>();
        Map<String, ConcurrentlyConsumer> consumerMap = applicationContext.getBeansOfType(ConcurrentlyConsumer.class);
        RocketMQProperties rocketMQProperties = applicationContext.getBean(RocketMQProperties.class);
        this.nameServers = rocketMQProperties.getNameServers();

        for (String key : consumerMap.keySet()) {
            ConcurrentlyConsumer consumer = consumerMap.get(key);
            MqConsumer mqConsumer = consumer.getClass().getAnnotation(MqConsumer.class);
            ConsumeMessageHook consumeMessageHook = (ConsumeMessageHook) applicationContext.getBean(mqConsumer.consumeMessageHook());
            if (mqConsumer != null) {
                DefaultMQPushConsumer defaultMQPushConsumer = (DefaultMQPushConsumer) this.createConsumer(mqConsumer.messageModel());
                //默认没有支持tag做区分
                defaultMQPushConsumer.subscribe(mqConsumer.topic(),"*");
                defaultMQPushConsumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(consumeMessageHook);
                this.registerConsumer(defaultMQPushConsumer);
                topicConsumerMap.put(mqConsumer.topic(),consumer);
                System.out.println("defaultMQPushConsumer is "+defaultMQPushConsumer.toString());
                defaultMQPushConsumer.start();
            }
        }
    }



    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public MQConsumer createConsumer(MessageModel messageModel) {
        DefaultMQPushConsumer mqConsumer = new DefaultMQPushConsumer();
        if(nameServers==null || nameServers==""){
            throw new RuntimeException("mq config has error,nameServers can not be null!");
        }
        mqConsumer.setNamesrvAddr(nameServers);
        mqConsumer.setMessageModel(messageModel);
        mqConsumer.setConsumerGroup(consumerConfig.getGroupName());
        mqConsumer.setConsumeThreadMax(consumerConfig.getMaxThreadNum());
        mqConsumer.setConsumeThreadMin(consumerConfig.getMinThreadNum());
        return mqConsumer;
    }

    /**
     * 核心用于设置消费模块程序
     */
    public void registerConsumer(DefaultMQPushConsumer mqPushConsumer){
        mqPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    ConcurrentlyConsumer concurrentlyConsumer = topicConsumerMap.get(messageExt.getTopic());
                    try {
                        //实际的消费
                        return concurrentlyConsumer.consume(messageExt);
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
    }
}
